package com.fstec.ks.master.queue;

import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.channels.FileChannel.MapMode;
import java.nio.charset.Charset;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.sinotn.SinotnLogger;
import com.sinotn.SinotnRuntime;

public class LocalFileReader {
	private static Logger LOG=LoggerFactory.getLogger("com.sinotn.ks");
	private static final Charset charset=Charset.forName("UTF-8");
	public static final String READ_META_FILE="read.m";
	// 读头文件16字节，
	//8字节[0-7]记录读取位置;
	//8字节[8-15]记录读取总记录数;
	private static final int READ_HEAD_SIZE=16;
	private RandomAccessFile headRaf;
	private FileChannel headChannel;
	private MappedByteBuffer writeMappedByteBuffer;
	private int fileSize;
	private boolean trace=LOG.isTraceEnabled();
	private RandomAccessFile readRaf;
	private FileChannel readChannel;
	private MappedByteBuffer readMappedByteBuffer;

	private DataFileRef dataRef;

	private AtomicLong readPos;
	private AtomicInteger readSize=new AtomicInteger(0);
	private AtomicLong readCount;
	private String dtname;
	private ByteBuffer readBuffer;
	private File folder;
	private String id;
	
	public LocalFileReader(File folder, int readBuffSize) {
		this.id=folder.getName();
		this.folder=folder;
		this.readBuffer=ByteBuffer.allocate(readBuffSize);
	}
	
	public long getReadCount(){
		if(this.readCount==null){
			return 0;
		}
		return this.readCount.get();
	}
	
	private void loadDataFile(long offset, long readCurPos) throws IOException {
		if(this.dataRef!=null){
			this.dataRef.close();
			this.dataRef=null;
		}
		File dtfile=new File(this.folder, ReceiverUtil.formatOffset(offset));
		if(!dtfile.exists()){
			throw new IOException("当前读位置"+this.readPos+"映射数据文件不存在"+dtfile.getAbsolutePath());
		}
		this.dataRef=new DataFileRef(dtfile, "r");
		this.dataRef.position(readCurPos);
		this.dtname=dtfile.getName();
		LOG.info("加载数据文件{}, readCount={}, readPos={}, readCurPos={}", dtfile.getName(), this.readCount.get(), this.readPos.get(), readCurPos);
	}

	public void init() {
		if(this.headRaf!=null) return;// 避免多次初始化
		File headWrite=new File(this.folder, LocalFileWriter.META_FILE);
		File headRead=new File(this.folder, READ_META_FILE);
		boolean ok=false;
		try{
			this.headRaf=new RandomAccessFile(headWrite, "r");
			this.headChannel=this.headRaf.getChannel();
			this.writeMappedByteBuffer=this.headChannel.map(MapMode.READ_ONLY, 0, LocalFileWriter.HEAD_SIZE);
			this.fileSize=this.writeMappedByteBuffer.getInt(16);

			this.readRaf=new RandomAccessFile(headRead, "rw");
			this.readChannel=this.readRaf.getChannel();
			this.readMappedByteBuffer=this.readChannel.map(MapMode.READ_WRITE, 0, READ_HEAD_SIZE);
			//8字节[0-7]记录读取位置;
			this.readPos=new AtomicLong(this.readMappedByteBuffer.getLong(0));
			//8字节[8-15]记录读取总记录数;
			this.readCount=new AtomicLong(this.readMappedByteBuffer.getLong(8));
			long offset=(this.readPos.get()/this.fileSize)*this.fileSize;
			this.loadDataFile(offset, this.readPos.get()%this.fileSize);
			LOG.info("消费者初始化[dir={},writePos={}]", this.folder.getName(),this.writeMappedByteBuffer.getLong(0));
			ok=true;
		}catch(Throwable e){
			throw new RuntimeException("初始化队列["+this.folder.getName()+"]消费者失败", e);
		}finally{
			if(!ok){
				this.close();
			}
		}
	}
	public void close() {
		if(this.writeMappedByteBuffer!=null){
			try{
				ByteBufferUtil.clean(this.writeMappedByteBuffer);
			}catch(Throwable e){}
		}
		if(this.headChannel!=null){
			try{
				this.headChannel.close();
			}catch(Throwable e){}
		}
		if(this.headRaf!=null){
			try{
				this.headRaf.close();
			}catch(Throwable e){}
		}

		if(this.readMappedByteBuffer!=null){
			try{
				this.readMappedByteBuffer.force();
			}catch(Throwable e){}
			try{
				ByteBufferUtil.clean(this.readMappedByteBuffer);
			}catch(Throwable e){}
		}
		if(this.readChannel!=null){
			try{
				this.readChannel.close();
			}catch(Throwable e){}
		}
		if(this.readRaf!=null){
			try{
				this.readRaf.close();
			}catch(Throwable e){}
		}

		if(this.dataRef!=null){
			try{
				this.dataRef.close();
			}catch(Throwable e){}
		}
	}
	/**
	 * 是否有未消费消息
	 * @return
	 * @auth <a href="mailto:xieqj@fstec.com">谢启进</a>
	 * @date 2022年5月15日 下午3:22:48
	 */
	public boolean hasNext() {
		final long writePos=this.writeMappedByteBuffer.getLong(0);
		return (this.readPos.get()<writePos);
	}
	
	/*
	 * 从数据文件读取指定字节数据
	 */
	private ByteBuffer readBuffer(int size) throws IOException{
		ByteBuffer buffer=this.readBuffer;
		if(size>this.readBuffer.capacity()){
			// 当前需要读取字节数据大于初始缓存大小，构造新的缓存对象
			buffer=ByteBuffer.allocate(size);
		}else{
			buffer.clear();
			buffer.limit(size);
		}
		// 先读字符串长度
		long lastPos=this.dataRef.position()+size;
		long dataSize=this.dataRef.size();
		if(trace){
			LOG.trace("readBuffer[size={},beginPos={},endPos={},dataSize={}]", size, this.dataRef.position(), lastPos,dataSize);
		}
		if(lastPos>dataSize){// 合并读取
			LOG.info("数据文件合并读取lastPos={}, fileSize={}, mergeSize={}", lastPos, dataSize, (lastPos-dataSize));
			// 当前文件不足size字节
			this.dataRef.read(buffer);
			// 加载下1个数据文件
			long newFilePos=((this.readPos.get()/this.fileSize)+1)*this.fileSize;
			this.loadDataFile(newFilePos, 0);
			this.dataRef.read(buffer);
		}else{
			this.dataRef.read(buffer);
		}
		this.readSize.addAndGet(size);
		return buffer;
	}

	public QueueMessage next() {
		final long writePos=this.writeMappedByteBuffer.getLong(0);
		int byteSize=this.readSize.get();
		this.readSize.set(0);
		if(byteSize>0){
			// 保存读取位置
			//8字节[0-7]记录读取位置;
			this.readMappedByteBuffer.putLong(0, this.readPos.addAndGet(byteSize));
			//8字节[8-15]记录读取总记录数;
			this.readMappedByteBuffer.putLong(8, this.readCount.incrementAndGet());
			if(trace){
				LOG.trace("当前处理{}字节消息，共{}条消息[queue={},dtfile={},readPos={},writePos={}]", byteSize, this.readCount.get(),this.id, this.dtname, this.readPos.get(), writePos);
			}
			byteSize=0;
		}
		// 消息队列文件写位置
		if(this.readPos.get()<writePos){// 当前读位置小于写位置表示有消息数据
			try{
				QueueMessage message=new QueueMessage();
				message.setPosition(this.readPos.get());
				String text;
				// 消息类型;
				ByteBuffer buffer=this.readBuffer(4);
				message.setType(buffer.getInt(0));
				// 时间戳timestamp
				buffer=this.readBuffer(8);
				message.setTimestamp(buffer.getLong(0));
				// 消息标识id
				buffer=this.readBuffer(1);
				byte lenByte=buffer.get(0);
				buffer=this.readBuffer(lenByte);
				buffer.flip();
				text=charset.decode(buffer).toString();
				message.setId(text);
				// 消息文本体bodyText
				buffer=this.readBuffer(4);
				int lenInt=buffer.getInt(0);
				buffer=this.readBuffer(lenInt);
				buffer.flip();
				text=charset.decode(buffer).toString();
				message.setBodyText(text);
				// 消息文件file
				buffer=this.readBuffer(4);
				lenInt=buffer.getInt(0);
				if(lenInt>0){
					buffer=this.readBuffer(lenInt);
					buffer.flip();
					text=charset.decode(buffer).toString();
					if(text.startsWith("home://")){
						text=SinotnRuntime.getHome().getAbsolutePath()+text.substring(7);
					}
					message.setFile(new File(text));
				}
				// 消息备注
				buffer=this.readBuffer(4);
				lenInt=buffer.getInt(0);
				if(lenInt>0){
					buffer=this.readBuffer(lenInt);
					buffer.flip();
					text=charset.decode(buffer).toString();
					message.setDescText(text);
				}
				
				return message;
			}catch(Throwable e){
				SinotnLogger.DEBUG.error("读取队列消息发生错误", e);
			}
		}
		return null;
	}
}
